package com.yomahub.liteflow.example.utils;

import cn.hutool.core.util.ObjectUtil;
import com.yomahub.liteflow.example.config.HdfsConfig;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.servlet.http.HttpServletResponse;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.URI;
import java.util.List;
import java.util.Map;

@Slf4j
@Data
@Component
public class HdfsUtil {
    @Autowired
    private HdfsConfig hdfsConfig;

    /**
     * @return FileSystem
     * @description 获取hdfs
     * @author yyyyjinying
     * @since 2023/6/5 11:14
     */
    public FileSystem getfileSystem() {
        Configuration conf = new Configuration();
        conf.set("dfs.client.use.datanode.hostname", "true");
        FileSystem fileSystem = null;

        try {
            fileSystem = FileSystem.get(new URI(hdfsConfig.getURI()), conf, hdfsConfig.getUsername());
        } catch (Exception e) {
            log.error("连接HDFS失败", e);
        }
        return fileSystem;
    }

    /**
     * @param String path
     * @return void
     * @description hdfs创建文件
     * @author yyyyjinying
     * @since 2023/6/5 15:03
     */
    public void mkdir(String path) {
        FileSystem fileSystem = getfileSystem();
        try {
            fileSystem.mkdirs(new Path(path));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param path 文件路径
     * @param bol  是否递归删除
     * @return void
     * @description 支持递给删除文件
     * @author yyyyjinying
     * @since 2023/6/5 16:10
     */
    public void delete(String path, boolean bol) {
        FileSystem fileSystem = getfileSystem();
        try {
            fileSystem.delete(new Path(path), bol);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param 1
     * @return void
     * @description 递归方式取得指定目录下的文件信息
     * @author yyyyjinying
     * @since 2023/6/5 16:23
     */
    public void getPathAllFiles() {
        FileSystem fileSystem = getfileSystem();
        // 递归方式取得指定目录下的文件信息
        FileStatus[] fileStatus = new FileStatus[0];
        try {
            fileStatus = fileSystem.listStatus(new Path("/"));
            // 输出取得的文件权限和文件名称  
            System.out.println("权限\t\t名称\t\t文件大小\t 修改时间");
            for (FileStatus fileStatus1 : fileStatus) {
                System.out.print(fileStatus1.getPermission() + "\t" + fileStatus1.getPath());
                System.out.print("\t" + fileStatus1.getLen());
                System.out.print("\t" + fileStatus1.getModificationTime());
                System.out.println();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param src 上传文件
     * @param dst hdfs地址
     * @return void
     * @description hdfs上传文件
     * @author yyyyjinying
     * @since 2023/6/5 11:13
     * Path src = new Path("E:\\note\\python-project\\bigpysparkone\\data\\sql\\people.txt");
     * Path dst = new Path("/input/javademo");
     */
    public void upFile(Path src, Path dst) {
        FileSystem fileSystem = getfileSystem();
        try {
            fileSystem.copyFromLocalFile(src, dst);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void moveLocalFile(Path src, Path dst) {
        FileSystem fileSystem = getfileSystem();
        try {
            fileSystem.moveFromLocalFile(src, dst);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param path         dfs
     * @param downloadPath 目标路径
     * @return void
     * @description 下载文件
     * @author yyyyjinying
     * @since 2023/6/5 16:45
     */
    public void downFile(String path, String downloadPath) {
        if (StringUtils.isEmpty(path) || StringUtils.isEmpty(downloadPath)) {
            return;
        }
        // 上传路径
        Path clientPath = new Path(path);
        // 目标路径
        Path serverPath = new Path(downloadPath);

        // 调用文件系统的文件复制方法，第一个参数是否删除原文件true为删除，默认为false
        try {
            FileSystem fileSystem = getfileSystem();
            fileSystem.copyToLocalFile(false, clientPath, serverPath);
        } catch (IOException e) {
            log.error(e.getMessage());
        }

    }


    /**
     * 将List<Map<String, String>>数据写入hdfs的csv文件中
     *
     * @param fieldNames
     * @param list
     * @param path
     */
    public void writeListMapToCSV(String fieldNames, List<Map<String, String>> list, String path) {
        FileSystem fileSystem = getfileSystem();
        getPathAllFiles();
        FSDataOutputStream outputStream = null;
        try {
            outputStream = fileSystem.create(new Path(path));
            // write csv header
            outputStream.writeBytes(fieldNames + "\n");
            // write csv data
            for (Map<String, String> recond : list) {
                String join = String.join(",", recond.values()) + "\n";
                outputStream.write(join.getBytes());
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            close(fileSystem, outputStream);
        }
    }

    /**
     * 页面下载csv
     * @param filename
     * @param path
     * @param response
     */
    public void downLoadToCsv(String filename, String path, HttpServletResponse response) {
        // 设置响应的内容类型为CSV文件
        response.setContentType("text/csv");
        // 设置响应的头信息，指定文件名
        response.setHeader("Content-Disposition", "attachment; filename="+filename);

        FileSystem fileSystem = getfileSystem();
        BufferedInputStream is = null;
        OutputStream os = null;
        try {
            is = new BufferedInputStream(fileSystem.open(new Path(path)));
            os = response.getOutputStream();
            IOUtils.copy(is, os);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }  finally {
            IOUtils.closeQuietly(os);
            IOUtils.closeQuietly(is);
        }
    }

    public void close(FileSystem fs, OutputStream os) {
        IOUtils.closeQuietly(os);
        try {
            if (ObjectUtil.isEmpty(fs)) {
                fs.close();
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
